package rabbitmq.publisher.confirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.contant.Contants;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;

public class PublisherConfirms {
    public static final Integer MESSAGE_COUNT = 500;
    static Connection createConnection() throws Exception{
        // 1.建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Contants.HOST);
        connectionFactory.setPort(Contants.PORT);
        connectionFactory.setUsername(Contants.USER_NAME);
        connectionFactory.setPassword(Contants.PASSWORD);
        connectionFactory.setVirtualHost(Contants.VIRTUAL_HOST);
        Connection connection = connectionFactory.newConnection();
        return connection;
    }

    /**
     * 异步确认
     * @throws Exception
     */
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {

        try (Connection connection = createConnection()){
            Channel channel = connection.createChannel();
            channel.confirmSelect();
            channel.queueDeclare(Contants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
            // 4. 监听confirm
            // 集合中存储的是未确认的消息ID
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());
            long start = System.currentTimeMillis();
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        confirmSeqNo.headSet(deliveryTag+1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        confirmSeqNo.headSet(deliveryTag+1).clear();
                    }else {
                        confirmSeqNo.remove(deliveryTag);
                    }
                }
                // 业务需要根据实际需求处理,比如消息重发
                // TODO
            });
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher_confirms "+i;
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("",Contants.PUBLISHER_CONFIRMS_QUEUE3,null,msg.getBytes());
                confirmSeqNo.add(seqNo);
            }
            while (!confirmSeqNo.isEmpty()){
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.printf("异步确认策略,消息条数: %d, 耗时: %d ms\n",MESSAGE_COUNT,end-start);

        }

    }

    /**
     * 批量确认
     */
    private static void publishingMessagesInBatches() throws Exception{
        try(Connection connection = createConnection()) {
            // 1. 开启信道
            Channel channel = connection.createChannel();
            // 2. 设置信道为confirm模式
            channel.confirmSelect();
            // 3. 声明队列
            channel.queueDeclare(Contants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);
            // 4. 发送消息,进行确认
            int batchSize = 100;
            int outstandMessageCount = 0;
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher_confirms "+i;
                channel.basicPublish("",Contants.PUBLISHER_CONFIRMS_QUEUE2,null,msg.getBytes());
                outstandMessageCount++;
                if (outstandMessageCount==batchSize){
                    channel.waitForConfirms(5000);
                    outstandMessageCount = 0;
                }
            }
            if (outstandMessageCount!=0){
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("批量确认策略,消息条数: %d, 耗时: %d ms\n",MESSAGE_COUNT,end-start);

        }
    }

    /**
     * 单独确认
     */
    private static void publishingMessagesIndividually() throws Exception {
        try (Connection connection = createConnection()){
            // 1.开启信道
            Channel channel = connection.createChannel();
            // 2.设置信道为confirm模式
            channel.confirmSelect();
            // 3.声明队列
            channel.queueDeclare(Contants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
            long start = System.currentTimeMillis();
            // 4.发送信息,等待确认
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "hello publisher_confirms "+i;
                channel.basicPublish("",Contants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes());
                // 等待确认
                channel.waitForConfirms(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("单独确认策略,消息条数: %d, 耗时: %d ms\n",MESSAGE_COUNT,end-start);
        }
    }

    /**
     * 确认发布
     * @param args
     */
    public static void main(String[] args) throws Exception {
        //Strategy #1: Publishing Messages Individually
        //单独确认
        publishingMessagesIndividually();
        //Strategy #2: Publishing Messages in Batches
        // 批量确认
        publishingMessagesInBatches();

        //Strategy #3: Handling Publisher Confirms Asynchronously
        // 异步确认
        handlingPublisherConfirmsAsynchronously();
    }


}
